package pbc.gx.SaveJson.kafka.service;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

@Service("saveJsonService")
public class SaveJsonService {
    public void save2Redis(RedisTemplate redisTemplate, List<ConsumerRecord<String, JSONObject>> records) {
        redisTemplate.executePipelined(new RedisCallback<Object>() {

            @Override
            public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
                for (ConsumerRecord<String, JSONObject> record : records) {
                    String key = record.key();
                    JSONObject value = record.value();
                    //System.out.println(key + "-" + value);
                    //手动提交offset
                    byte[] keyB = redisTemplate.getKeySerializer().serialize(key);
                    Map<byte[], byte[]> hvalue = new HashMap<>();
                    Iterator<String> iter = value.keySet().iterator();
                    while (iter.hasNext()) {
                        Object hkey = iter.next();
                        byte[] hkeyB = redisTemplate.getHashKeySerializer().serialize(hkey);
                        byte[] hvalueB = redisTemplate.getHashValueSerializer().serialize(value.get(hkey));
                        hvalue.put(hkeyB, hvalueB);
                    }
                    redisConnection.hMSet(keyB, hvalue);
                    redisConnection.close();

                }
                return null;
            }
        });

    }


    public void save2Mysql(JdbcTemplate jdbcTemplate, List<ConsumerRecord<String, JSONObject>> records) {
        String tableName = records.get(0).key().split(":")[0];
        Set<String> keySet = records.get(0).value().keySet();
        String sql = "insert into " + tableName + "(";
        String valueStr = "(";
        for (String keyStr : keySet) {
            sql += keyStr + ",";
            valueStr += "?,";
        }
        sql = sql.substring(0, sql.lastIndexOf(",")) + ") values " + valueStr.substring(0, valueStr.lastIndexOf(",")) + ")";
        jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() {
            @Override
            public void setValues(PreparedStatement preparedStatement, int i) throws SQLException {
                int j = 1;
                for (String keyStr : keySet) {
                    preparedStatement.setObject(j, records.get(i).value().getObject(keyStr, Object.class));
                    j++;
                }
            }

            @Override
            public int getBatchSize() {
                return records.size();
            }
        });
    }
}
